#include "app_pool.h"

// 消息队列存储的数据类型
typedef struct
{
    void (*task)(void *);
    void *data;
} app_pool_task_t;

typedef struct
{
    pthread_t *threads;
    int pool_size;

    mqd_t mq;
    char *mq_name;
} app_pool_t;

static app_pool_t my_pool;

void *app_pool_thread(void *arg);
gate_state_t app_pool_init(int pool_size)
{
    my_pool.pool_size = pool_size;
    my_pool.mq_name = "/pool";

    // 先创建一个消息队列
    struct mq_attr attr;
    attr.mq_flags = 0;
    attr.mq_maxmsg = pool_size;
    attr.mq_msgsize = sizeof(app_pool_task_t);

    my_pool.mq = mq_open(my_pool.mq_name, O_CREAT | O_RDWR, 0666, &attr);
    if (my_pool.mq == -1)
    {
        log_error("消息队列创建失败");
        return GATE_ERROR;
    }
    log_info("消息队列创建成功");

    // 分配内存, 存储线程id
    my_pool.threads = malloc(sizeof(pthread_t) * pool_size);

    for (size_t i = 0; i < pool_size; i++)
    {
        int r = pthread_create(&my_pool.threads[i], NULL, app_pool_thread, NULL);
        if (r == -1)
        {
            log_error("pthread_create error");
            return GATE_ERROR;
        }
    }
    log_info("线程池初始化 ok");
    return GATE_OK;
}

void app_pool_deinit(void)
{
    for (size_t i = 0; i < my_pool.pool_size; i++)
    {
        pthread_cancel(my_pool.threads[i]);
    }

    free(my_pool.threads);

    mq_close(my_pool.mq);
    mq_unlink(my_pool.mq_name);

    log_info("线程池销毁 ok");
}

void *app_pool_thread(void *arg)
{
    while (1)
    {
        app_pool_task_t task;
        // 从消息队列中取出一个函数(任务), 然后执行
        ssize_t len = mq_receive(my_pool.mq, (char *)&task, sizeof(app_pool_task_t), NULL);
        if (len > 0 && task.task)
        {
            task.task(task.data);
        }
    }
}

gate_state_t app_pool_add_task(void (*task)(void *), void *data)
{
    if (task == NULL)
    {
        log_error("要执行的任务不能时null");
        return GATE_ERROR;
    }

    app_pool_task_t task_info = {
        .task = task,
        .data = data};
    int r = mq_send(my_pool.mq, (char *)&task_info, sizeof(app_pool_task_t), 0);

    if (r == -1)
    {
        log_error("给线程池添加任务失败");
        return GATE_ERROR;
    }
    log_debug("给线程池添加任务成功");
    return GATE_OK;
}
